package com.gmall.app.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.gmall.bean.TableProcess;
import com.gmall.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import org.apache.flink.configuration.Configuration;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;


public class TableProcessFunction extends BroadcastProcessFunction<JSONObject,String,JSONObject> {
    private OutputTag<JSONObject> objectOutputTag;
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private Connection connection;

    public TableProcessFunction(OutputTag<JSONObject> objectOutputTag, MapStateDescriptor<String, TableProcess> mapStateDescriptor){
        this.objectOutputTag=objectOutputTag;
        this.mapStateDescriptor=mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws SQLException, ClassNotFoundException {
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        connection= DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    @Override
    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext readOnlyContext, Collector<JSONObject> collector) throws Exception {
        //获取状态过滤数据
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);
        String key = jsonObject.getString("tableName") + "-" + jsonObject.getString("type");
        TableProcess tableProcess = broadcastState.get(key);

        if (tableProcess != null){
            //过滤字段
            JSONObject data = jsonObject.getJSONObject("after");
            filterColumn(data,tableProcess.getSinkColumns());

            //3.分流
            //将输出表/主题信息写入Value
            jsonObject.put("sinkTable", tableProcess.getSinkTable());
            String sinkType = tableProcess.getSinkType();
            if (TableProcess.SINK_TYPE_KAFKA.equals(sinkType)) {
                //Kafka数据,写入主流
                collector.collect(jsonObject);
            } else if (TableProcess.SINK_TYPE_HBASE.equals(sinkType)) {
                //HBase数据,写入侧输出流
                readOnlyContext.output(objectOutputTag, jsonObject);
            }

        } else {
            System.out.println("该组合Key：" + key + "不存在！");

        }
    }

    private void filterColumn(JSONObject data, String sinkColumns) {
        String[] fields = sinkColumns.split(",");
        List<String> columns = Arrays.asList(fields);

        data.entrySet().removeIf(next -> !columns.contains(next.getKey()));
    }

    @Override
    public void processBroadcastElement(String s, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
        //获取并解析数据
        JSONObject jsonObject = JSON.parseObject(s);
        String data = jsonObject.getString("after");
        TableProcess tableProcess = JSON.parseObject(data, TableProcess.class);

        //建表
        if (TableProcess.SINK_TYPE_HBASE.equals(tableProcess.getSinkType())){
           checkTable(tableProcess.getSinkTable(),
                   tableProcess.getSinkPk(),
                   tableProcess.getSinkExtend(),
                   tableProcess.getSinkColumns());
        }

        // 写入状态，广播出去
        BroadcastState<String, TableProcess> broadcastState = context.getBroadcastState(mapStateDescriptor);
        String key = tableProcess.getSourceTable() + "-" + tableProcess.getOperateType();
        broadcastState.put(key,tableProcess);
    }

    //建表语句 create table if not exists db.tn(id varchar parimary key, tm_name varchar ...)
    private void checkTable(String sinkTable, String sinkPk, String sinkExtend, String sinkColumns) {
        PreparedStatement preparedStatement=null;

        try{
            if (sinkPk == null){
                sinkPk = "id";
            }
            if (sinkExtend == null){
                sinkExtend="";
            }

            StringBuffer createTableSql = new StringBuffer("create table if not exists ")
                    .append(GmallConfig.HBASE_SCHEMA)
                    .append(".")
                    .append(sinkTable)
                    .append("(");

            //分隔符分开后放入String集合
            String[] fields = sinkColumns.split(",");

            for (int i = 0; i<fields.length;i++){
                String feild = fields[i];

                //判断是否为主键
                if (sinkPk.equals(feild)){
                    createTableSql.append(feild).append(" varchar primary key ");
                }else {
                    createTableSql.append(feild).append(" varchar ");
                }

                //判断是否是最后一个字段，如果不是，则添加","
                if (i<fields.length-1){
                    createTableSql.append(",");
                }
            }

            createTableSql.append(")").append(sinkExtend);

            //打印建表语句
            System.out.println(createTableSql);

            //预编译SQL
            PreparedStatement preparedStatement1 = connection.prepareStatement(createTableSql.toString());
            //启动
            preparedStatement1.execute();
        //修改报错信息
        } catch (SQLException e) {
            throw new RuntimeException("Phoenix表" + sinkTable + "建表失败");
        } finally {
            if (preparedStatement != null){
                try{
                    preparedStatement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
